package com.king.first.cdc;

import com.ververica.cdc.connectors.mysql.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.DebeziumSourceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class FlinkCDC {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
//
//        env.enableCheckpointing(5000L);
//
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
//
//        env.setStateBackend(new FsStateBackend("hdfs://hadoop01:8020/flink/test/cdc-ck"));

        DebeziumSourceFunction<String> mysqlFunction = MySqlSource.<String>builder()
                .hostname("rm-8vbgy7j9l5170r6mwto.mysql.zhangbei.rds.aliyuncs.com")
                .port(3306)
                .username("wudianlong")
                .password("Wyf25YswB6QV45RdfhgVB6U3")
                .databaseList("yidian_ceshi")
                .tableList("yidian_ceshi.tb_ma_log")
                .startupOptions(StartupOptions.latest())
                .deserializer(new CDCWithCustomerSchema())
                .build();

        DataStreamSource<String> dataStreamSource = env.addSource(mysqlFunction);


        dataStreamSource.print("cdc:");

        env.execute("cdc-test:");
    }
}
